package com.atguigu.api

import org.apache.flink.api.common.functions.{MapFunction, ReduceFunction, RichMapFunction}
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

/**
 * @description: xxx
 * @time: 2020/6/19 18:14
 * @author: baojinlong
 **/
object TransformTest2 {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    val inputStreamFromFile: DataStream[String] = environment.readTextFile("E:/qj_codes/big-data/FlinkTutorial/src/main/resources/sensor.data")

    // 基本转换操作
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      })
    // keyBy之后所有的统计都是根据当前key值来的
    //.keyBy("id")
    //.keyBy(data => data.id)
    // 分组滚动聚合
    val resultData: DataStream[SensorReading] = dataStream.keyBy(new MyIdSelector())
      //.sum("temperature")
      //.min("temperature")
      // min只会显示最小值和第一个其它字段,而minBy是当前key最小值对应的所有数据
      // .minBy("temperature")
      //  取时间最大,温度最小
      //      .reduce((curRes, newData) => {
      //        SensorReading(curRes.id, curRes.timestamp.max(newData.timestamp), curRes.temperature.min(newData.temperature))
      //      })
      .reduce(new MyReduce)

    // 分流操作,将流从逻辑上分开
    val splitStream: SplitStream[SensorReading] = resultData.split(data => {
      if (data.temperature > 30) {
        Seq("high")
      } else {
        Seq("low")
      }
    })

    // 分别提出流
    val highStream: DataStream[SensorReading] = splitStream.select("high")
    val lowStream: DataStream[SensorReading] = splitStream.select("low")
    val allStream: DataStream[SensorReading] = splitStream.select("high", "low")

    // 打印数据
    highStream.print("hight")
    lowStream.print("low")
    allStream.print("all")

    // 打印数据
    // resultData.print("resultData")

    // 合流
    val warningStream: DataStream[(String, Double, String)] = highStream.map(data => {
      (data.id, data.temperature, "high temp warning")
    })

    val connectedStream: ConnectedStreams[(String, Double, String), SensorReading] = warningStream.connect(lowStream)
    environment.execute("transform test job")
    val finalResultStream: DataStream[Product] = connectedStream.map(warningData => {
      (warningData._1, warningData._2, "high temp warning")
    },
      lowData => (lowData.id, "normal")
    )
    finalResultStream.print("finalResultStream")

    // union操作
    val unionResult: DataStream[SensorReading] = highStream.union(lowStream, allStream)
    unionResult.print("unionData")

  }

}


// 使用直接new MyMap
class MyMap extends MapFunction[SensorReading, (String, Double)] {
  override def map(value: SensorReading): (String, Double) = {
    (value.id, value.timestamp)
  }
}

class MyRichMap extends RichMapFunction[SensorReading, Int] {
  override def open(parameters: Configuration): Unit = {
    println("打开啦")
  }

  println(getRuntimeContext.getIndexOfThisSubtask)

  override def map(in: SensorReading): Int = {
    in.timestamp.toInt
  }

  override def close(): Unit = {
    println("关闭啦")
  }


}